//package com.ibm.cps.servlet;
//
//import java.util.ArrayList;
//import java.util.Collection;
//import java.util.HashSet;
//import java.util.List;
//import java.util.Map;
//import java.util.Set;
//
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import com.fasterxml.jackson.databind.JsonNode;
//import com.fasterxml.jackson.databind.node.ArrayNode;
//import com.fasterxml.jackson.databind.node.ObjectNode;
//import com.ibm.cps.kafka.KafkaProducer;
//import com.ibm.cps.kafka.KafkaTopicConstructor;
//import com.ibm.cps.message.MessageFields;
//import com.ibm.cps.message.MetadataTopologySort;
//import com.ibm.cps.message.validation.ValidatorUtil;
//import com.ibm.cps.model.AdapaterFactory;
//import com.ibm.cps.model.DataSourceMonitor;
//import com.ibm.cps.model.MetadataAdapter;
//import com.ibm.cps.model.ProcessorMetadtaValidator;
//import com.ibm.cps.model.ServletUtil;
//import com.ibm.cps.processors.ProcessorsGroup;
//import com.ibm.factories.PublishProcessorFactory;
//import com.ibm.interfaces.IPersistency;
//import com.ibm.mongo.MongoPersistency;
//import com.ibm.util.ErrorCode;
//import com.ibm.util.JSONUtility;
//import com.ibm.util.LocalConfig;
//import com.ibm.util.exception.CPSException;
//
///**
// * implement Processor servlet logic in this class
// *
// * @author Jingjing Wang
// *
// *         2015-7-16
// */
//public class ProcessorsImplementation implements ServletTemplate {
//
//	private Logger logger;
//	private KafkaProducer producer;
//	private IPersistency persistency;
//	private ProcessorMetadtaValidator processorValidator;
//
//	public ProcessorsImplementation() throws CPSException {
//		logger = LoggerFactory.getLogger(this.getClass());
//		producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
//		logger.info("Connect to kafka " + LocalConfig.KAFKA_BROKERS);
//		persistency = MongoPersistency.getInstace();
//		logger.info("Connect to persistency " + LocalConfig.MONGO_CONNECT_STRING + ", database is "
//				+ LocalConfig.PERSISTENCY_DATABASE);
//		processorValidator = ProcessorMetadtaValidator.getInstance();
//	}
//
//	/**
//	 * validate processor/group metadata
//	 *
//	 * @param object
//	 * @param type
//	 * @param source
//	 * @param tenantid
//	 * @return
//	 * @throws CPSException
//	 */
//	private String validateMetadata(ObjectNode object, String type, String source, String tenantid)
//			throws CPSException {
//		String processorid = AssignIdFactory.getProcessorid(type, object);
//		boolean success = processorValidator.validate(tenantid, processorid, object, source);
//		if (!success) {
//			return null;
//		}
//		return constructRetMsg(source, processorid, type);
//	}
//
//	/**
//	 * construct return message according to different conditions
//	 *
//	 * @param source
//	 * @param processorid
//	 * @param type
//	 * @return
//	 */
//	private String constructRetMsg(String source, String processorid, String type) {
//		if (ProcessorsGroup.isGroupType(type)) {
//			ObjectNode retNode = JSONUtility.newObjectNode();
//			retNode.put("id", processorid);
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retNode);
//		}
//		if (source.equals(MessageFields.SQL) || source.equals(MessageFields.NODERED)) {
//			return null;
//		} else {
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, processorid);
//		}
//	}
//
//	@Override
//	public String doPost(Map<String, String> key2Value, String input) throws CPSException {
//		ArrayNode retArray = JSONUtility.newArrayNode();
//		ArrayList<ObjectNode> sendMsg = new ArrayList<ObjectNode>();
//		ArrayList<ObjectNode> groups = new ArrayList<ObjectNode>();
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		String source = key2Value.get(MessageFields.SOURCE);
//		if (source == null) {
//			source = "rest";
//		}
//		try {
//			DataSourceMonitor monitor = new DataSourceMonitor();
//			PublishProcessorFactory publishFactory = PublishProcessorFactory.getInstance();
//			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//			input = adapter.adapt2Json(input, source, tenantid);
//			if (input == null) {
//				return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Null input metadata.");
//			}
//
//			logger.info("Adapter output: " + input);
//			ArrayNode objects = JSONUtility.getArrayFromString(input);
//
//			Collection<String> existedMetadatas = persistency.getProcessors(tenantid, null, null, null);
//			ArrayNode arrayNode = JSONUtility.newArrayNode(existedMetadatas);
//			if (arrayNode != null) {
//				arrayNode.addAll(objects);
//				MetadataTopologySort.getSortedMetadatas(arrayNode);
//			}
//			List<ObjectNode> metadatas = MetadataTopologySort.getSortedMetadatas(objects);
//			for (ObjectNode object : metadatas) {
//				String type = JSONUtility.GetString(MessageFields.TYPE, object);
//				String result = validateMetadata(object, type, source, tenantid);
//				if (!ProcessorsGroup.isGroupType(type)) {
//					sendMsg.add(object);
//				} else {
//					groups.add(object);
//				}
//				if (result == null) {
//					continue;
//				}
//				ObjectNode node = JSONUtility.fromObjectString(result);
//				retArray.add(node);
//			}
//
//			// if all metadata pass the validation, then send all of them to
//			// persistency and kafka
//			for (ObjectNode metadata : sendMsg) {
//				producer.send(KafkaTopicConstructor.getMetadataAddTopic(tenantid), metadata.toString().getBytes());
//				persistency.putProcessor(metadata);
//				logger.info("Put message into persistency:" + metadata);
//			}
//
//			for (ObjectNode groupMetadata : groups) {
//				persistency.putGroupMetadata(groupMetadata);
//				// persistency.putMetadata(tenantid, groupMetadata.toString());
//			}
//
//			return retArray.toString();
//		} catch (CPSException e) {
//			for (ObjectNode tmpObject : sendMsg) {
//				processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
//			}
//			for (ObjectNode tmpObject : groups) {
//				processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
//			}
//			throw e;
//		}
//
//	}
//
//	@Override
//	public String doGet(Map<String, String> key2Value) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		String source = key2Value.get(MessageFields.SOURCE);
//		if (source == null) {
//			source = "rest";
//		}
//		DataSourceMonitor monitor = new DataSourceMonitor();
//		PublishProcessorFactory publishFactory = PublishProcessorFactory.getInstance();
//		MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//		Collection<String> metadatas = persistency.getProcessors(tenantid, key2Value.get(MessageFields.PROCESSORID),
//				key2Value.get(MessageFields.TYPE), key2Value.get(MessageFields.GROUPID));
//		// Collection<String> metadatas =
//		// getMetadatas(key2Value.get(MessageFields.SOURCE), tenantid,
//		// key2Value.get(MessageFields.TYPE),
//		// key2Value.get(MessageFields.GROUPID),
//		// key2Value.get(MessageFields.PROCESSORID));
//		Collection<String> allMetadata = persistency.getProcessors(tenantid, null, null, null);
//		// Collection<String> allMetadata =
//		// persistency.getMetadataByUser(tenantid);
//		Collection<String> result = adapter.getJsons(allMetadata, metadatas);
//		return ServletUtil.constructCollectionJson(result);
//	}
//
//	@Override
//	public String doDelete(Map<String, String> key2Value, String input) throws CPSException {
//		throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "This kind of delete method is unsupported.");
//	}
//
//	private String deleteProcessors(String tenantid, Collection<String> deletedMetadata, String source)
//			throws CPSException {
//		String retMessage;
//		ArrayNode deletedIds = JSONUtility.newArrayNode();
//		if (deletedMetadata != null && deletedMetadata.size() >= 1) {
//			for (String deleted : deletedMetadata) {
//				ArrayNode tmp = processorValidator.deleteProcessor(tenantid, deleted, source);
//				if (tmp != null && tmp.size() >= 1) {
//					for (int index = 0; index < tmp.size(); index++) {
//						deletedIds.add(tmp.get(index));
//					}
//					producer.send(KafkaTopicConstructor.getMetadataDeleteTopic(tenantid), deleted.getBytes());
//				}
//			}
//			if (deletedIds.size() >= 1) {
//				retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE,
//						deletedIds.toString() + " have been deleted.");
//			} else {
//				retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing has been deleted.");
//			}
//		} else {
//			retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
//		}
//		return retMessage;
//	}
//
//	/**
//	 * delete sql group
//	 *
//	 * @param input
//	 * @throws CPSException
//	 */
//	private void deleteGroupProcessors(String tenantid, String input, String source) throws CPSException {
//		JsonNode arrayNode = JSONUtility.getArrayFromString(input);
//		Set<String> groupids = new HashSet<String>();
//		for (int i = 0; i < arrayNode.size(); i++) {
//			ObjectNode element = (ObjectNode) arrayNode.get(i);
//			String groupid = JSONUtility.GetString(MessageFields.GROUPID, element);
//			if (groupid != null) {
//				groupids.add(groupid);
//			}
//		}
//
//		if (groupids.size() < 1) {
//			throw new CPSException(ErrorCode.ERROR_FIELD_TYPE, "Can't get groupid from the adapter output " + input);
//		}
//		for (String groupid : groupids) {
//			Collection<String> deletedProcessors = persistency.getProcessors(tenantid, null, null, groupid);
//			Collection<String> groupMetadata = persistency.getGroupMetadata(tenantid, groupid);
//			// Collection<String> deletedProcessors =
//			// persistency.getGroupProcessors(tenantid, groupid);
//			// String groupMetadata = persistency.getGroupById(tenantid,
//			// groupid);
//			if (groupMetadata == null) {
//				logger.error("Group metadta of groupid " + groupid + " should be existed.");
//			} else {
//				deletedProcessors.addAll(groupMetadata);
//			}
//			deleteProcessors(tenantid, deletedProcessors, source);
//		}
//	}
//
//	@Override
//	public String doDelete(Map<String, String> key2Value) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//		Collection<String> deletedMetadata = ProcessorsGetFactory.getMetadatasAndGroups(key2Value.get(MessageFields.SOURCE),
//				tenantid, key2Value.get(MessageFields.TYPE), key2Value.get(MessageFields.GROUPID),
//				key2Value.get(MessageFields.PROCESSORID));
//		String source = key2Value.get(MessageFields.SOURCE);
//		if (deletedMetadata != null) {
//			return deleteProcessors(tenantid, deletedMetadata, source);
//		} else {
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
//		}
//	}
//
//	@Override
//	public String doPut(Map<String, String> key2Value, String input) throws CPSException {
//		String tenantid = key2Value.get(MessageFields.TENANTID);
//		ValidatorUtil.checkNotNull(MessageFields.TENANTID, tenantid);
//
//		String source = key2Value.get("source");
//		if (source == null) {
//			source = "rest";
//		}
//
//		// if (!source.equals(MessageFields.SQL)) {
//		// throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, source
//		// + " can't do the put operation.");
//		// }
//		DataSourceMonitor monitor = new DataSourceMonitor();
//		PublishProcessorFactory publishFactory = PublishProcessorFactory.getInstance();
//		MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//		String adaptInput = adapter.adapt2Json(input, source, tenantid);
//
//		if (adaptInput == null) {
//			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Null input metadata.");
//		} else {
//			logger.info("Adapter output: " + adaptInput);
//			deleteGroupProcessors(tenantid, adaptInput, source);
//			return doPost(key2Value, input);
//		}
//	}
//}
